package main

import (
	"fmt"
	"log"
	"os"
	"time"

	"github.com/Shopify/sarama"
	"github.com/spf13/viper"
)

// 测试用例修改
func init() {
	viper.SetConfigFile("../conf.yaml") //定义需要读取文件的路径
	err := viper.ReadInConfig()         //尝试读取
	if err != nil {
		fmt.Println("配置文件读取错误！~err:", err)
		return
	}
	addr = viper.GetString("kafka.addr")
	topic = viper.GetString("kafka.topic")
}

var (
	addr  string //地址
	topic string //主题
)

//同步发送信息
/* func main() {

	config := sarama.NewConfig()
	config.Version = sarama.V3_3_1_0
	config.Producer.Idempotent = true                // 开启幂等性，确保消息不会重复发送
	config.Net.MaxOpenRequests = 1                   // 设置同时打开的最大请求数为5，以控制网络传输的并发性能
	config.Producer.Return.Errors = true             // 当消息发送失败时，将错误返回给生产者
	config.Producer.Return.Successes = true          // 当消息发送成功时，将成功消息返回给生产者
	config.Producer.RequiredAcks = sarama.WaitForAll // 等待所有副本都收到消息后，再发送成功消息给生产者
	//config.Producer.Partitioner = sarama.NewHashPartitioner // 设置消息的分区策略为哈希分区  NewRandomPartitioner
	//config.Metadata.Retry.Max = 3                          //表示在获取元数据的过程中最多重试 3 次
	//config.Metadata.Retry.Backoff = 250 * time.Millisecond //表示在每次重试之间等待 250 毫秒

	sarama.Logger = log.New(os.Stdout, "[Sarama] ", log.LstdFlags)

	producer, err := sarama.NewSyncProducer([]string{"192.168.80.200:9092"}, config)
	if err != nil {
		panic(err)
	}
	i := 0
	for {
		i++
		msg := WechatMessage{
			Content:    fmt.Sprintf("msg %d", i),
			From:       fmt.Sprintf("from %d", i),
			CreateTime: time.Now().Format("2006-01-02  15:04:05"),
		}
		data, _ := json.Marshal(msg)
		p, o, err := producer.SendMessage(&sarama.ProducerMessage{
			// Topic:     "wechat_msg_p",
			Topic: "test",
			Value: sarama.ByteEncoder(data),
			//Key: sarama.StringEncoder("小张"),
			//Partition: int32(i) % 3,
		})
		fmt.Println(p, o, err)
		time.Sleep(time.Second)
	}
} */

// 异步发送信息
func main() {
	config := sarama.NewConfig()
	config.Version = sarama.V3_3_1_0
	config.Producer.Return.Errors = true    // 当消息发送失败时，将错误返回给生产者
	config.Producer.Return.Successes = true // 当消息发送成功时，将成功消息返回给生产者
	config.Producer.Partitioner = sarama.NewHashPartitioner
	sarama.Logger = log.New(os.Stdout, "[Sarama] ", log.LstdFlags)

	// 创建一个新的异步生产者
	producer, err := sarama.NewAsyncProducer([]string{addr}, config)
	if err != nil {
		fmt.Println("Error creating producer:", err)
		return
	}
	defer func() {
		if err := producer.Close(); err != nil {
			fmt.Println("Error closing producer:", err)
		}
	}()

	// 定义一个信道，用于接收消息发送结果
	successes := producer.Successes()
	// 等待所有消息发送结果
	go func() {
		for {
			select {
			case success := <-successes:
				fmt.Printf("Message sent to 分区是 %d at 偏移量 %d\n", success.Partition, success.Offset)
			case err := <-producer.Errors():
				fmt.Println("Error sending message:", err)
				// case <-time.After(time.Second):
				// 	fmt.Println("Timeout sending message")
			}
		}
	}()
	for {
		time.Sleep(time.Second)
		// 准备要发送的消息
		messages := sarama.ProducerMessage{
			Topic: topic,
			Value: sarama.StringEncoder("你好呀！"),
		}
		// 发送多条消息
		producer.Input() <- &messages

	}

}

type WechatMessage struct {
	Content    string `json:"content"`
	From       string `json:"from"`
	CreateTime string `json:"create_time"`
}
